package cn.trigram.mq;

import cn.trigram.ddo.entity.MqTransactionEntity;
import cn.trigram.mapper.MqTransactionMapper;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.stereotype.Component;

/**
 * service01 mq生产
 *
 * @see <a href="https://github.com/apache/rocketmq/tree/master/docs/cn">最佳实践</a>
 */
@Slf4j
@Component
@Setter(onMethod_ = {@Autowired})
public class Service01MqProducer {

  private RocketMQTemplate rocketMQTemplate;

  private MqTransactionMapper transactionMapper;

  @Setter
  private String topic = "local-service01-topic";

  public void oneWay(String tags, String key, Object payload) {

    String destination = wrapDestination(tags);
    try {
      MessageHeaders   messageHeaders   = wrapHeaders(key);
      MessageConverter messageConverter = this.rocketMQTemplate.getMessageConverter();
      Message<?>       message          = messageConverter.toMessage(payload, messageHeaders);
      this.rocketMQTemplate.sendOneWay(destination, message);
    } catch (Exception e) {
//      一般出错后重试，可以考虑异步任务实现
      log.error("[{}]消息发送失败", destination, e);
    }
  }

  public void transactionSend(String tags, String key, Object payload) {

    String destination = wrapDestination(tags);
    try {
      String              transactionId     = UUID.randomUUID().toString();
      MqTransactionEntity transactionEntity = new MqTransactionEntity();
      transactionEntity.setId(UUID.randomUUID().toString());
      transactionEntity.setTransactionId(transactionId);
      transactionEntity.setLog("插入负载:" + payload);
      transactionEntity.setStatus("init");
      transactionMapper.insertSelective(transactionEntity);
      Map<String, Object> headerMap = new HashMap<>();
      headerMap.put(RocketMQHeaders.KEYS, key);
      headerMap.put(RocketMQHeaders.TRANSACTION_ID, transactionId);
      headerMap.put("uuid", key);

      MessageHeaders messageHeaders = new MessageHeaders(headerMap);

      MessageConverter messageConverter = this.rocketMQTemplate.getMessageConverter();
      Message<?>       message          = messageConverter.toMessage(payload, messageHeaders);
      TransactionSendResult sendResult = this.rocketMQTemplate.sendMessageInTransaction(
          destination, message,
          transactionId
      );
      logSendResult(destination, sendResult);
      LocalTransactionState transactionState = sendResult.getLocalTransactionState();
      if (LocalTransactionState.COMMIT_MESSAGE == transactionState) {
        log.info("事务消息提交成功");
      } else if (LocalTransactionState.ROLLBACK_MESSAGE == transactionState) {
        log.error("事务消息提交失败，已经回滚");
      } else if (LocalTransactionState.UNKNOW == transactionState) {
        log.warn("事务消息状态未知");
      }
    } catch (Exception e) {
//      一般出错后重试，可以考虑异步任务实现
      log.error("[{}]事务消息发送失败", destination, e);
    }
  }

  public void syncSend(String tags, String key, Object payload) {

    String destination = wrapDestination(tags);
    try {
      MessageHeaders   messageHeaders   = wrapHeaders(key);
      MessageConverter messageConverter = this.rocketMQTemplate.getMessageConverter();
      Message<?>       message          = messageConverter.toMessage(payload, messageHeaders);
      SendResult       sendResult       = this.rocketMQTemplate.syncSend(destination, message);
      logSendResult(destination, sendResult);
    } catch (Exception e) {
//      一般出错后重试，可以考虑异步任务实现
      log.error("[{}]消息发送失败", destination, e);
    }
  }

  public void syncSendOrderly(String tags, String key, Object payload) {

    String destination = wrapDestination(tags);
    try {
      MessageHeaders   messageHeaders   = wrapHeaders(key);
      MessageConverter messageConverter = this.rocketMQTemplate.getMessageConverter();
      Message<?>       message          = messageConverter.toMessage(payload, messageHeaders);
      SendResult sendResult = this.rocketMQTemplate.syncSendOrderly(
          destination, message, key);
      logSendResult(destination, sendResult);
    } catch (Exception e) {
//      一般出错后重试，可以考虑异步任务实现
      log.error("[{}]消息发送失败", destination, e);
    }
  }

  public void aysncSend(String tags, String key, Object payload) {

    String           destination      = wrapDestination(tags);
    MessageHeaders   messageHeaders   = wrapHeaders(key);
    MessageConverter messageConverter = this.rocketMQTemplate.getMessageConverter();
    Message<?>       message          = messageConverter.toMessage(payload, messageHeaders);
    this.rocketMQTemplate.asyncSend(
        destination,
        message,
        new SendCallback() {

          @Override
          public void onSuccess(
              SendResult sendResult
          ) {

            logSendResult(destination, sendResult);
          }

          @Override
          public void onException(
              Throwable e
          ) {
//      一般出错后重试，可以考虑异步任务实现
            log.error("[{}]消息发送失败", destination, e);
          }
        }
    );

  }

  private void logSendResult(String destination, SendResult result) {
    /*https://github.com/apache/rocketmq/blob/master/docs/cn/best_practice.md#3-%E6%97%A5%E5%BF%97%E7%9A%84%E6%89%93%E5%8D%B0*/
    SendStatus status = result.getSendStatus();
    if (status == SendStatus.SEND_OK) {
      log.info("[{}]消息发送成功, result：[{}]", destination, result);
    } else if (status == SendStatus.FLUSH_DISK_TIMEOUT) {
      log.info("[{}]消息发送成功但是服务器刷盘超时, result：[{}]", destination, result);
    } else if (status == SendStatus.FLUSH_SLAVE_TIMEOUT) {
      log.info("[{}]消息发送成功，但是服务器同步到Slave时超时, result：[{}]", destination, result);
    } else if (status == SendStatus.SLAVE_NOT_AVAILABLE) {
      log.info("[{}]消息发送成功，但是此时Slave不可用, result：[{}]", destination, result);
    }
  }

  private String wrapDestination(String tags) {

    return topic + ":" + tags;
  }

  /**
   * 包装请求头
   *
   * @see RocketMQUtil#convertToRocketMessage
   */
  private MessageHeaders wrapHeaders(String... headers) {

    Map<String, Object> headerMap = new HashMap<>();
    String              key       = findHeaders(headers, 0);
    headerMap.put(RocketMQHeaders.KEYS, key);
    return new MessageHeaders(headerMap);
  }

  private String findHeaders(String[] headers, int i) {

    return Objects.nonNull(headers) && i < headers.length ? headers[i] : null;
  }

}
